package com.szjz.elasticsearch.eshelper.base.wrapper;

import com.szjz.elasticsearch.eshelper.base.BaseESEntity;
import com.szjz.elasticsearch.eshelper.base.wrapper.service.DayWrapperService;
import com.szjz.elasticsearch.eshelper.service.ESService;
import com.szjz.elasticsearch.eshelper.wrapper.ESQueryWrapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.core.CountRequest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author 史博
 * @desc 按天分库
 * @date 2021/2/28 12:28 周日
 */
@Slf4j
@Service
public class DayWrapperServiceImpl<T extends BaseESEntity> extends MonthWrapperServiceImpl<T> implements DayWrapperService<T> {
	@Autowired
	private ESService esService;

	/*====================================================================*/
	/*================================= SAVE =============================*/
	/*====================================================================*/

	/**
	 * 保存or更新
	 *
	 * @param t
	 */
	@Override
	public Boolean saveOrUpdateAcDay(T t) {
		if (t == null) {
			log.warn("【ES saveOrUpdateAcDay】t can not be null");
			return false;
		}
		if (StringUtils.isEmpty(t.getId())) {
			log.warn("【ES saveOrUpdateAcDay】t'id can not be null");
			return false;
		}
		return saveOrUpdateBatchAcDay(Collections.singletonList(t));
	}

	/**
	 * 批量 保存or更新数据
	 * 按照数据的创建时间来分片
	 * ts中的数据创建时间必须是同一月的
	 * docId = _id
	 *
	 * @param ts 数据集合
	 */
	@SneakyThrows
	@Override
	public Boolean saveOrUpdateBatchAcDay(List<T> ts) {
		if (CollectionUtils.isEmpty(ts)) {
			log.warn("【ES saveOrUpdateBatchAcDay】ts list Can not be empty !");
			return false;
		}
		List<T> crtTimes = ts.stream().filter(t -> t.getCrtTime() == null).collect(Collectors.toList());
		if (!CollectionUtils.isEmpty(crtTimes)) {
			log.warn("【ES saveOrUpdateBatchAcDay】crtTimes : {}", crtTimes);
			log.warn("【ES saveOrUpdateBatchAcDay】There are elements in the collection whose crtTime time is null!");
			throw new RuntimeException("There are elements in the collection whose crtTime time is null!");
		}
		List<T> ids = ts.stream().filter(t -> t.getId() == null).collect(Collectors.toList());
		if (!CollectionUtils.isEmpty(ids)) {
			log.warn("【ES saveOrUpdateBatchAcDay】ids : {}", ids);
			log.warn("【ES saveOrUpdateBatchAcDay】The attribute 'id' in the collection cannot be null!");
			throw new RuntimeException("The attribute 'id' in the collection cannot be null!");
		}
		Map<Integer, List<T>> collect = ts.stream().collect(Collectors.groupingBy(item -> item.getCrtTime().getDay()));
		for (Map.Entry<Integer, List<T>> entry : collect.entrySet()) {
			// Integer key = entry.getKey();
			List<T> list = entry.getValue();
			// log.info("key:{} size ：{} crtTime: {} list:{} ", key, list.size(), list.get(0).getCrtTime(), list.stream().map(t -> t.getCrtTime()).collect(Collectors.toList()));
			BulkRequest bulkRequest = new BulkRequest();
			BulkResponse bulkResponse = null;
			String indexName = getIndexName(list.get(0).getCrtTime());
			for (T t : list) {
				t.setDocId(t.getId());
				if (t.getUpdTime() == null) {
					t.setUpdTime(Calendar.getInstance().getTime());
				}
				IndexRequest indexRequest = new IndexRequest(indexName)
					.source(getSourceMap(t));
				UpdateRequest updateRequest = new UpdateRequest(indexName, t.getDocId())
					.doc(indexRequest)
					.docAsUpsert(true);
				bulkRequest.add(updateRequest);
			}
			bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
			assert bulkResponse != null;
			if (bulkResponse.hasFailures()) {
				if (!esService.indexExist(indexName)) {
					Boolean isSuccess = esService.createIndex(indexName, getEntityClass());
					log.info("【ES saveOrUpdateBatchAcDay】索引库 {} 不存在, 重新创建{}", indexName, isSuccess ? "成功！" : "失败！");
					if (isSuccess) {
						saveOrUpdateBatchAcDay(ts);
					} else {
						throw new RuntimeException("【ES saveOrUpdateBatchAcDay】索引库 " + indexName + " 创建失败");
					}
				} else {
					// saveOrUpdateBatchAcDay(ts);
					log.error("【ES saveOrUpdateBatch】保存失败原因:{}", bulkResponse.buildFailureMessage());
					// log.error("【ES saveOrUpdateBatchAcDay】失败！重新保存 indexName: {} size: {} data：{}", indexName, list.size(), list.get(0));
				}
				return false;
			} else {
				log.debug("【ES saveOrUpdateBatchAcDay】成功！indexName: {} size: {}", indexName, list.size());
			}
		}
		return true;
	}

	/*====================================================================*/
	/*================================ SELECT ============================*/
	/*====================================================================*/


	/**
	 * 通过包装类查询按天分片的索引库集合
	 *
	 * @param wrapper 将要查询的条件封装到wrapper中
	 * @param date    索引库对应的分片时间
	 * @return 某一天的数据集合
	 */
	@Override
	public List<T> selectAcDay(ESQueryWrapper<T> wrapper, Date date) {
		return selectAcDay(wrapper, date, SIZE);
	}

	/**
	 * 通过包装类查询按天分片的索引库集合
	 *
	 * @param wrapper   将要查询的条件封装到wrapper中
	 * @param startDate 开始时间
	 * @param endDate   结束时间
	 * @return 某一时间段的数据集合
	 */
	@Override
	public List<T> selectAcDay(ESQueryWrapper<T> wrapper, Date startDate, Date endDate) {
		return selectAcDay(wrapper, startDate, endDate, SIZE);
	}

	/**
	 * 通过包装类查询按天分片的索引库集合
	 *
	 * @param wrapper 将要查询的条件封装到wrapper中
	 * @param date    索引库对应的分片时间
	 * @param size    返回结果集的最大数量
	 * @return 某一天的数据集合
	 */
	@Override
	public List<T> selectAcDay(ESQueryWrapper<T> wrapper, Date date, Integer size) {
		SearchRequest searchRequest = new SearchRequest(getIndexName(date));
		return getSelectTs(wrapper, searchRequest, size);
	}


	/**
	 * 通过包装类查询按天分片的索引库集合
	 *
	 * @param wrapper   将要查询的条件封装到wrapper中
	 * @param startDate 开始时间
	 * @param endDate   结束时间
	 * @param size      返回结果集的最大数量
	 * @return 某一时间段的数据集合
	 */
	@Override
	public List<T> selectAcDay(ESQueryWrapper<T> wrapper, Date startDate, Date endDate, Integer size) {
		SearchRequest searchRequest = getSearchRequestAcDay(startDate, endDate);
		return getSelectTs(wrapper, searchRequest, size);
	}

	/**
	 * count
	 * 可用于非等查询
	 *
	 * @param wrapper 将要查询的条件封装到ESQueryWrapper中
	 * @param date    指定要查询的索引库
	 * @return
	 */
	@Override
	public Long countAcDay(ESQueryWrapper<T> wrapper, Date date) {
		CountRequest countRequest = new CountRequest(getIndexName(date));
		return getCount(countRequest, wrapper);
	}

	/**
	 * count
	 * 可用于非等查询
	 *
	 * @param wrapper   将要查询的条件封装到ESQueryWrapper中
	 * @param startDate 指定要查询的索引库的开始时间
	 * @param endDate   指定要查询的索引库的开始时间
	 * @return
	 */
	@Override
	public Long countAcDay(ESQueryWrapper<T> wrapper, Date startDate, Date endDate) {
		CountRequest countRequest = new CountRequest(getIndexNamesAcDay(startDate, endDate));
		return getCount(countRequest, wrapper);
	}
}
